package com.carrot.mq.mqserver;

import com.carrot.mq.common.Consumer;
import com.carrot.mq.common.MqException;
import com.carrot.mq.mqserver.datacenter.DiskDataCenter;
import com.carrot.mq.mqserver.datacenter.MemoryDataCenter;
import com.carrot.mq.mqserver.core.*;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author 兴趣使然黄小黄
 * @version 1.0
 * @date 2023/8/12 20:20
 * 虚拟主机, 不同虚拟主机之间的内容互不影响
 * 每个虚拟主机管理各自: 交换机, 队列, 绑定, 消息数据
 * 并提供 api 供上层调用
 */
public class VirtualHost {
    // 虚拟主机名
    private String virtualHostName;
    // 管理内存数据
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    // 管理硬盘数据
    private DiskDataCenter diskDataCenter = new DiskDataCenter();
    // 转发规则
    private Router router = new Router();
    private ConsumerManager consumerManager = new ConsumerManager(this);
    // 操作交换机的锁对象
    private final Object exchangeLocker = new Object();
    // 操作队列的锁对象
    private final Object queueLocker = new Object();


    public VirtualHost(String name) {
        this.virtualHostName = name;
        // 对于硬盘管理, 需要先建库建表和目录
        diskDataCenter.init();
        // 需要针对硬盘的数据恢复到内存
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (IOException | MqException | ClassNotFoundException e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 恢复内存数据失败!");
        }
    }

    public String getVirtualHostName() {
        return virtualHostName;
    }

    public MemoryDataCenter getMemoryDataCenter() {
        return memoryDataCenter;
    }

    public DiskDataCenter getDiskDataCenter() {
        return diskDataCenter;
    }

    /**
     * 创建交换机(交换机从属于虚拟主机, 交换机的名字 = 虚拟主机的名字 + 交换机的名字)
     * 交换机存在直接返回, 交换机不存在则创建一个交换机
     * @param exchangeName 交换机名称
     * @param exchangeType 交换机类型
     * @param durable 持久化
     * @param autoDelete 自动删除
     * @param arguments 自定义配置
     * @return true 创建成功, false 创建失败
     */
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable, boolean autoDelete,
                                   Map<String, Object> arguments) {
        // 交换机名字加上虚拟主机前缀
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                // 查询当前交换机是否存在, 直接通过内存查询
                Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                if (existsExchange != null) {
                    // 交换机已经存在
                    System.out.println("[VirtualHost] 交换机已经存在! exchangeName=" + exchangeName);
                    return true;
                }
                // 创建交换机
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(exchangeType);
                exchange.setAutoDelete(autoDelete);
                exchange.setDurable(durable);
                exchange.setArguments(arguments);
                // 先写硬盘后写内存, 如果硬盘写入失败, 就不需要写入内存
                if (durable) {
                    // 交换机写入硬盘
                    diskDataCenter.insertExchange(exchange);
                }
                // 交换机写入内存
                memoryDataCenter.insertExchange(exchange);
                System.out.println("[VirtualHost] 交换机创建完成! exchangeName=" + exchangeName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 交换机创建失败! exchangeName=" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 删除交换机
     * @param exchangeName 交换机名称
     * @return 删除成功 true
     */
    public boolean exchangeDelete(String exchangeName) {
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                // 找到交换机
                Exchange toDelete = memoryDataCenter.getExchange(exchangeName);
                if (toDelete == null) {
                    throw new MqException("[VirtualHost] 交换机不存在无法删除!");
                }
                // 删除硬盘上的数据
                if (toDelete.isDurable()) {
                    diskDataCenter.deleteExchange(exchangeName);
                }
                // 删除内存中的交换机数据
                memoryDataCenter.deleteExchange(exchangeName);
                System.out.println("[VirtualHost] 交换机删除成功! exchangeName=" + exchangeName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 交换机删除失败! exchangeName=" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 创建队列
     * @param queueName 队列名
     * @param durable 是否持久化
     * @param exclusive 是否独有
     * @param autoDelete 是否自动删除
     * @param arguments 自定义配置
     * @return 创建成功 true
     */
    public boolean queueDeclare(String queueName, boolean durable, boolean exclusive, boolean autoDelete,
                               Map<String, Object> arguments) {
        // 队列的名字拼接上虚拟主机的名字
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                // 判定队列是否存在
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                if (existsQueue != null) {
                    System.out.println("[VirtualHost] 队列已经存在! queueName=" + queueName);
                    return true;
                }
                // 创建队列对象
                MSGQueue queue = new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(arguments);
                // 写入硬盘
                if (durable) {
                    diskDataCenter.insertQueue(queue);
                }
                // 写入内存
                memoryDataCenter.insertQueue(queue);
                System.out.println("[VirtualHost] 队列创建成功! queueName=" + queueName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 队列创建失败! queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 删除队列
     * @param queueName 队列名
     * @return 删除成功 true
     */
    public boolean queueDelete(String queueName) {
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                // 根据队列名查询队列对象
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if (queue == null) {
                    throw new MqException("[VirtualHost] 队列不存在无法删除! queueName=" + queueName);
                }
                // 删除硬盘数据
                if (queue.isDurable()) {
                    diskDataCenter.deleteQueue(queueName);
                }
                // 删除内存数据
                memoryDataCenter.deleteQueue(queueName);
                System.out.println("[VirtualHost] 删除队列成功! queueName" + queueName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 删除队列失败! queueName" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 实现交换机与队列的绑定
     * @param queueName 队列名称
     * @param exchangeName 交换机名称
     * @param bindingKey 绑定 Key
     * @return 绑定成功 true
     */
    public boolean queueBind(String queueName, String exchangeName, String bindingKey) {
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    // 判定当前的绑定是否已经存在, 若已经存在则为失败
                    Binding existsBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (existsBinding != null) {
                        throw new MqException("[VirtualHost] binding 已经存在! queueName=" + queueName
                                + ", exchangeName=" + exchangeName);
                    }
                    // 验证 bindingKey 是否合法
                    if (!router.checkBindingKey(bindingKey)) {
                        throw new MqException("[VirtualHost] bindingKey 非法! bindingKey=" + bindingKey);
                    }
                    // 创建 Binding 对象
                    Binding binding = new Binding();
                    binding.setQueueName(queueName);
                    binding.setExchangeName(exchangeName);
                    binding.setBindingKey(bindingKey);
                    // 检查对应的交换机和队列是否存在, 如果有任意一个不存在则说明该绑定是无法创建的
                    MSGQueue queue = memoryDataCenter.getQueue(queueName);
                    if (queue == null) {
                        throw new MqException("[VirtualHost] 队列不存在! queueName=" + queueName);
                    }
                    Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                    if (exchange == null) {
                        throw new MqException("[VirtualHost] 交换机不存在! exchangeName=" + exchangeName);
                    }
                    // 写入硬盘
                    if(queue.isDurable() && exchange.isDurable()) {
                        diskDataCenter.insertBinding(binding);
                    }
                    // 写入内存
                    memoryDataCenter.insertBinding(binding);
                }
            }
            System.out.println("[VirtualHost] 绑定创建成功! queueName=" + queueName
                    + ", exchangeName=" + exchangeName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] queueBind 失败!");
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 删除绑定
     * @param queueName 队列名称
     * @param exchangeName 交换机名称
     * @return 删除成功 true
     */
    public boolean queueUnbind(String queueName, String exchangeName) {
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    // 检查绑定是否存在, 如果不存在则不需要删除
                    Binding binding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (binding == null) {
                        throw new MqException("[VirtualHost] 绑定不存在无法删除! exchangeName=" + exchangeName
                                + ", queueName=" + queueName);
                    }
                    // 以下被注释的代码有一个 bug , 如果用户先删除了交换机或者先删除了队列, 而后就无法删除绑定
                    // 可以参考 mysql 外键一样, 在删除交换机和队列的时候检查是否存在相关的绑定, 如果存在则不允许删除交换机或者队列
                    // 但是需要重新组织队列和绑定之间的数据结构
                    // 笔者采用直接删除的方式, 因为 binding 是存储在数据库中的, 其 delete 就是一条 sql,
                    // 而对于内存中的 binding, 在删除之前已经进行了非空校验
//            // 检查队列和交换机是否存在
//            MSGQueue queue = memoryDataCenter.getQueue(queueName);
//            if (queue == null) {
//                throw new MqException("[VirtualHost] 对应的队列不存在! queueName=" + queueName);
//            }
//            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
//            if (exchange == null) {
//                throw new MqException("[VirtualHost] 对应的交换机不存在! exchangeName=" + exchangeName);
//            }
                    // 删除硬盘上的数据
                    diskDataCenter.deleteBinding(binding);
                    // 删除内存上的数据
                    memoryDataCenter.deleteBinding(binding);
                }
            }
            System.out.println("[VirtualHost] 绑定删除成功! queueName=" + queueName
                    + ", exchangeName=" + exchangeName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 绑定删除失败!");
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 发送消息到指定的队列中
     * @param exchangeName 交换机名称
     * @param routingKey 对于 TopicExchange: routingKey 为消息上带有的内容, 与 bindingKey 匹配
     *                   对于 DirectExchange: routingKey 表示要转发的队列名
     *                   对于 FanoutExchange: routingKey 无意义
     * @param basicProperties 消息的属性
     * @param body 消息的正文
     * @return 发送消息成功 true
     */
    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties, byte[] body) {
        try {
            // 转发交换机的名字
            exchangeName = virtualHostName + exchangeName;
            // 检查 routingKey 是否合法
            if (!router.checkRoutingKey(routingKey)) {
                throw new MqException("[VirtualHost] routingKey 非法! routingKey=" + routingKey);
            }
            // 查找交换机对象
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if (exchange == null) {
                throw new MqException("[VirtualHost] 交换机不存在! exchangeName=" + exchangeName);
            }
            // 判定交换机类型, 根据不同交换机类型进行对应的消息发送
            if (exchange.getType() == ExchangeType.DIRECT) {
                // 直接交换机, routingKey 作为队列的名字, 直接把消息写入指定的队列中
                // 可以无视绑定关系, 弱化了绑定的限制
                String queueName = virtualHostName + routingKey;
                // 构造消息对象
                Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                // 查找队列对应的对象
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if (queue == null) {
                    throw new MqException("[VirtualHost] 队列不存在! queueName=" + queueName);
                }
                // 向队列写入消息
                sendMessage(queue, message);
            } else {
                // fanout 与 topic 交换机
                // 找到该交换机关联的所有绑定
                ConcurrentHashMap<String, Binding> bindingsMap = memoryDataCenter.getBindings(exchangeName);
                for (Map.Entry<String, Binding> entry : bindingsMap.entrySet()) {
                    // <queueName, Binding>
                    // 获取绑定对象, 判定对应的队列是否存在
                    Binding binding = entry.getValue();
                    MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
                    if (queue == null) {
                        System.out.println("[VirtualHost] basicPublish 发送消息时, 队列不存在! queueName="
                                + binding.getQueueName());
                        continue;
                    }
                    // 构造消息对象
                    Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                    // 判定该消息是否能转发给该队列
                    // 1) fanout 交换机, 则发送给该交换机绑定的所有队列
                    // 2) topic 交换机, 则只转发给 bindingKey 与 routingKey 匹配的
                    if (!router.route(exchange.getType(), binding, message)) {
                        // 不能转发
                        continue;
                    }
                    // 转发消息给队列
                    sendMessage(queue, message);
                }
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 消息发送失败!");
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 向指定队列发送消息, 将消息写入硬盘和内存上
     * @param queue 队列
     * @param message 消息
     */
    private void sendMessage(MSGQueue queue, Message message) throws IOException, MqException, InterruptedException {
        int deliverMode = message.getDeliverMode();
        if (deliverMode == 2) {
            // 持久化, 写入硬盘
            diskDataCenter.sendMessage(queue, message);
        }
        // 写入内存
        memoryDataCenter.sendMessage(queue, message);
        // 通知消费者可以消费消息
        consumerManager.notifyConsume(queue.getName());
    }

    /**
     * 订阅消息
     * 添加一个队列的订阅者, 当队列收到消息后, 就会把消息推送给对应的订阅者
     * @param consumerTag 消费者的身份标识
     * @param queueName 队列名
     * @param autoAck 消息被消费后的应答方式, true 自动应答, false 手动应答
     * @param consumer 回调函数, 使用函数式接口, 后续调用 basicConsume 并且传递实参的时候可以写作 lambda
     * @return 订阅成功 true
     */
    public boolean basicConsume(String consumerTag, String queueName, boolean autoAck, Consumer consumer) {
        // 构造 ConsumerEnv 对象, 找到对应的队列, 将 Consumer 对象添加到该队列中
        queueName = virtualHostName + queueName;
        try {
            consumerManager.addConsumer(consumerTag, queueName, autoAck, consumer);
            System.out.println("[VirtualHost] basicConsume 成功! queueName=" + queueName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] basicConsume 失败! queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 手动确认应答
     * @param queueName 队列名
     * @param messageId 消息 id
     * @return 应答成功 true
     */
    public boolean basicAck(String queueName, String messageId) {
        queueName = virtualHostName + queueName;
        try {
            Message message = memoryDataCenter.getMessage(messageId);
            if (message == null) {
                throw new MqException("[VirtualHost] 要确认的消息不存在! messageId=" + messageId);
            }
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if (queue == null) {
                throw new MqException("[VirtualHost] 要确认的队列不存在! queueName=" + queueName);
            }
            // 删除硬盘上的数据
            if (message.getDeliverMode() == 2) {
                diskDataCenter.deleteMessage(queue, message);
            }
            // 删除消息中心的数据
            memoryDataCenter.removeMessage(messageId);
            // 删除待确认集合中的数据
            memoryDataCenter.removeMessageWaitAck(queueName, messageId);
            System.out.println("[VirtualHost] basicAck 成功! 消息被成功确认! queueName=" + queueName
                    + ", messageId=" + messageId);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] basicAck 失败! 消息确认失败! queueName=" + queueName
                    + ", messageId=" + messageId);
            e.printStackTrace();
            return false;
        }
    }
}
